package com.itbac.redis_lock.lock;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Component;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * Redis 分布式锁
 * @author: BacHe
 * @email: 1218585258@qq.com
 * @Date: 2020/8/28 0:33
 */
@Component
public class RedisLockImpl implements RedisLock {

    //前缀
    private String keyPrefix = "lock_";
    //默认超时时间：毫秒
    private long defaultTime = 20000;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    //尝试获取锁
    @Override
    public boolean tryLock(String key) {
        boolean flag =  stringRedisTemplate.execute(new RedisCallback<Boolean>() {
            @Override
            public Boolean doInRedis(RedisConnection redisConnection) throws DataAccessException {
                String value = "lock";
                //setnx 方法
                return redisConnection.set((keyPrefix + key).getBytes(), value.getBytes(),
                        Expiration.milliseconds(defaultTime), RedisStringCommands.SetOption.SET_IF_ABSENT);
            }
        });
        System.out.println("1.RedisLockImpl.tryLock:" + flag);
        return flag;
    }
    //加锁。
    @Override
    public void lock(String key) {
        //先加本地锁。
        // 本地锁使得一个节点，只有一个连接去监听Redis，减少Redis连接压力。
//     synchronized (key){

             //尝试获取锁，失败，循环等待。
             while (!tryLock(key)){
                 //睡眠
//                             try {
//                                 Thread.sleep(1000L);
//                             } catch (InterruptedException e) {
//                                 e.printStackTrace();
//                             }

                 //订阅Redis的锁释放通道

                 Long execute = stringRedisTemplate.execute(new RedisCallback<Long>() {
                     @Override
                     public Long doInRedis(RedisConnection redisConnection) throws DataAccessException {
                        //倒计数器
                         CountDownLatch waiter = new CountDownLatch(1);
                                 //subscribe订阅消息
                         redisConnection.subscribe((message, pattern) -> {
                             System.out.println("收到消息：" + message);
                             //监听器的行为： 减一
                             waiter.countDown();
                             //监听的通道
                         }, ("lock_notice:" + key).getBytes());
                         //阻塞等待
                         try {
                             waiter.await();
                         } catch (Exception e) {

                         }
                         return 0L;
                     }
                 });
                 System.out.println(Thread.currentThread().getName() + "循环tryLock");
                 //再循环。去抢锁。
             }

//     }

    }

    @Override
    public void lock(String key, long time) {

        synchronized (key){
            //尝试获取锁，失败，循环等待。
            while (!tryLock(key)){
                //睡眠
                //            try {
                //                Thread.sleep(100L);
                //            } catch (InterruptedException e) {
                //                e.printStackTrace();
                //            }

                //订阅Redis的锁释放通道
                stringRedisTemplate.execute(new RedisCallback<Long>() {
                    @Override
                    public Long doInRedis(RedisConnection redisConnection) throws DataAccessException {
                        //倒计数器
                        CountDownLatch waiter = new CountDownLatch(1);
                        redisConnection.subscribe((message, pattern) -> {
                            //监听器的行为： 减一
                            waiter.countDown();
                            //监听的通道
                        }, ("lock_notice:" + key).getBytes());
                        //阻塞等待
                        try {
                            //等待一定毫秒
                            waiter.await(time, TimeUnit.MILLISECONDS);
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                        return null;
                    }
                });
                //退出循环。
                break;
            }
        }

    }
    //释放锁
    @Override
    public void unlock(String key) {
        //删除掉key
        stringRedisTemplate.delete(keyPrefix + key);
        //通知其他资源抢锁。
        stringRedisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection redisConnection) throws DataAccessException {
                //锁释放的通知 通道
                Long publish = redisConnection.publish(("lock_notice:" + key).getBytes(), (Thread.currentThread().getName()+"锁释放").getBytes());
                return publish;
            }
        });
    }
}
